package com.example.bigdata.spark.文件读写

import java.sql.DriverManager

import org.apache.hadoop.io.LongWritable
import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.parsing.json.JSON

object ReadAndWriteSparkDemo {
    def main(args: Array[String]): Unit = {
        val inputPath = args(0)
        val outputPath = args(1)
        println("inputPath : " +args(0)+"\noutputPath : " +args(1))
        val conf = SparkContextInit()
        //读取文件(conf,hdfsFile)
        //val hdfsFile = "hdfs://sz-pg-dc-hadoop-005.tendcloud.com:9000/tmp/chapter4_3_6.txt"
        //读取hdfs文件(conf,hdfsFile)
        readAndWrite(conf,inputPath,outputPath)
        /*// 读取csv文件
        val csvfile = "input/user.csv"
        读取CSV文件(conf,csvfile)

        //  读取单文件
        val textFile = "input/Chapter4.3.1.txt"
        读取文件文件(conf,textFile)

        //  读取多文件
        val textFiles = "input/Chapter*.txt"
        读取文件文件(conf,textFiles)

        // 读取csv文件
        val csvfile = "input/user.csv"
        读取CSV文件(conf,csvfile)

        // 读取tsv文件
        val tsvFile = "input/user.tsv"
        读取TSV文件(conf,tsvFile)

        //读取文件(conf,hdfsFile)
        val hdfsFile = "hdfs://172.23.6.178:9820/user/hadoop/chapter4_3_6.txt"
        读取hdfs文件(conf,hdfsFile)

        //  读取Json文件
        val jsonFile = "input/info.json"
        读取JSON文件(conf,jsonFile)

        //  文件存储
        saveAs普通文件(conf,"output/普通文件/")

        saveAsCSV_TSV(conf)
        */



    }

    def SparkContextInit(): SparkConf ={
        // 初始化 sparkconf 对象，设置基本任务参数
        val conf = new SparkConf()
                // 设置目标Master 通信地址
                .setMaster("yarn")
                //.setMaster("local[1]")
                //设置任务名称
                .setAppName("读取hdfs文件-jingwei.shi")
        //实例化 SparkContext ，Spark 的对外接口负责用户与Spark内部的交互®通信
        return conf
    }

    def 读取文件(conf:SparkConf,textFile:String): Unit ={
        conf.setAppName("单文件读取")
        val sc = new SparkContext(conf)

        // 读取文本文件
        val inputTextFile = sc.textFile(textFile)
        println(inputTextFile.collect.mkString(","))
        sc.stop()
    }
    def 读取hdfs文件(conf:SparkConf,hdfsTextFile:String): Unit ={
        conf.setAppName("读取hdfs文件")
        val sc = new SparkContext(conf)
        println("xxxx")
        val inputTextFile = sc.newAPIHadoopFile(hdfsTextFile)
        println("xxxx")
        val result = inputTextFile.map(_._2.toString).collect
        println(result.mkString("\n"))
        sc.stop()
    }
    def 读取JSON文件(conf:SparkConf,textFile:String): Unit ={
        conf.setAppName("Json文件读取")
        val sc = new SparkContext(conf)
        val inputTextFile = sc.textFile(textFile)
        // 读取文本文件
        val content = inputTextFile.map(JSON.parseFull)
        println(content.collect.mkString("\t"))
        content.foreach({
            // 读取每条数据并返回 为 map 类型 ，获取address 的数据  {"name": "Thomas","age": "20","address": ["通讯1","通讯2"]}
            case Some(map:Map[String,Any]) => println(map.get("address"))
            case None => println("无效的Json")
            case _ => println("其他异常")
        })
        sc.stop()
    }
    def 读取CSV文件(conf:SparkConf,textFile:String): Unit ={
        conf.setAppName("读取CSV文件")
        val sc = new SparkContext(conf)
        val csvFile = sc.textFile(textFile)
                .flatMap(_.split(","))
                .collect
        csvFile.foreach(println)
        sc.stop()
    }
    def 读取TSV文件(conf:SparkConf,textFile:String): Unit ={
        conf.setAppName("读取TSV文件")
        val sc = new SparkContext(conf)
        val tsvFile = sc.textFile(textFile)
                .flatMap(_.split("\t"))
                .collect
        tsvFile.foreach(println)
        sc.stop()
    }
    def saveAs普通文件(conf:SparkConf,savePath:String): Unit ={
        conf.setAppName("文件存储")
        val sc = new SparkContext(conf)
        // parallelize 方法的 第二个参数 numSlices 用于设置分区个数，每个分区对应一个文件
        val rddData = sc.parallelize(Array(
            ("one",1),
            ("two",2),
            ("three",3)
        ) , 1)
        rddData.saveAsTextFile(savePath)
        sc.stop()
    }
    def readAndWrite(conf:SparkConf,inputPath:String,outputPath:String): Unit ={
        val sc = new SparkContext(conf)
        val df1 = sc.textFile(inputPath)
        df1.foreach(println)
        df1.repartition(1).saveAsTextFile(outputPath)
    }
    def saveAsCSV_TSV(conf:SparkConf): Unit ={
        conf.setAppName("saveAsCSV_TSV")
        val sc = new SparkContext(conf)
        // 模拟个人信息
        val array = Array("Thonmas",18,"male","65kg","180cm")
        // csv 格式保存
        val csvRDD = sc.parallelize((Array (array.mkString(","))),1)
        csvRDD.saveAsTextFile("output/csv")
        // tsv 格式保存
        val tsvRDD = sc.parallelize((Array (array.mkString("\t"))),1)
        tsvRDD.saveAsTextFile("output/tsv")
    }
}
